package com.hexinfo.dmpro.monitor.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hexinfo.dmpro.common.base.BaseService;
import com.hexinfo.dmpro.common.base.Page;
import com.hexinfo.dmpro.common.base.PageResult;
import com.hexinfo.dmpro.common.cache.Cache;
import com.hexinfo.dmpro.common.cache.CacheFactory;
import com.hexinfo.dmpro.monitor.dao.ClusterWhDao;
import com.hexinfo.dmpro.monitor.model.ClusterHostModel;
import com.hexinfo.dmpro.monitor.model.base.MonitorBaseModel;
import com.hexinfo.dmpro.monitor.model.base.MonitorReq;
import com.hexinfo.dmpro.monitor.util.ConvertUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.File;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;

/**
 * @author peng
 * @date 2022/2/28 14:32
 */
@Service
public class ClusterWhService extends BaseService {

    private static Cache cache;
    @Autowired
    private ClusterWhDao clusterWhDao;
    @Autowired
    private ClusterHostService clusterHostService;
    @Autowired
    private ClusterHostRoleService clusterHostRoleService;
    private String shellPath;
    private final String[] serviceType = {"YARN", "IMPALA", "HDFS", "HIVE"};
    @Autowired
    private ClusteAllRolesService clusteAllRolesService;
    @Autowired
    private ConvertUtil convertUtil;


    public ClusterWhService(CacheFactory cacheFactory) {
        cache = cacheFactory.createCache();
    }

    //集群信息管理-分页列表
    public PageResult selCmInfo(Page page, MonitorBaseModel monitorBaseModel) {
        return selectPage(clusterWhDao, "selCmInfo", page, monitorBaseModel);
    }

    //通过集群编码、子集群编码，获取集群信息
    public MonitorBaseModel queryCmInfo(String cmFlag,String clusterFlag) {
        return clusterWhDao.queryCmInfo(cmFlag,clusterFlag);
    }

    //集群信息管理-添加（涉及删除redis缓存）
    public int saveCmInfo(Map map) {
        String cmFlag = map.get("cmFlag").toString();
        if (cmFlag.contains("KAFKA")) {
            map.put("monitorFlag", "2");
        } else {
            map.put("monitorFlag", "1");
        }

        //先清空数据 在新增
        clusterWhDao.delCmInfo(map);
        int i = clusterWhDao.saveCmInfo(map);
        //更新集群信息时 删除 redis缓存数据
        cache.delLike("monitor_info:");
        return i;
    }

    //集群信息管理-删除（涉及删除redis缓存）
    public int delCmInfo(Set<String> ids) {
        //先清空数据 在新增
        int i = clusterWhDao.delCmInfoById(ids);
        cache.delLike("monitor_info:");
        return i;
    }

    public int saveHostInfo(Map map) {
        //先清空数据 在新增
        clusterWhDao.delHostInfo(map);
        //修改数据时 清空缓存
        cache.delLike("monitor_info:");
        return clusterWhDao.saveHostInfo(map);
    }

    /**
     * KAFKA 集群 cmFlag 必须包含KAFKA 关键字   monitorFlag 字段值为 2 表示有效的kafka集群
     *
     * @param map
     * @return
     */
    public int upMonitorFlag(Map map) {
        String cmFlag = map.get("cmFlag").toString();
        String monitorFlag = map.get("monitorFlag").toString();
        if (cmFlag.contains("KAFKA") && "1".equals(monitorFlag)) {
            map.put("monitorFlag", "2");
        }
        //修改数据时 清空缓存  暂时不放入缓存
        cache.delLike("monitor_info:");
        return clusterWhDao.upMonitorFlag(map);
    }

    //集群信息管理-集群配置刷新（涉及redis缓存）
    public Map refreshCluster(List<MonitorBaseModel> monitorBaseModels) {
        String dataTime = LocalDateTime.now().toString().substring(0, 19).replace("T", " ");

        Map<String, Integer> returnMap = new HashMap<>();
        int totalNum = 0;

        for (MonitorBaseModel monitorBaseModel : monitorBaseModels) {
            MonitorReq monitorReq = new MonitorReq();
            monitorReq.setTaskTime(dataTime);
            monitorReq.setUrl(monitorBaseModel.getUrl());
            monitorReq.setOtherParam(monitorBaseModel.getOtherParam());
            List<ClusterHostModel> hostInfos = clusterHostService.getHostInfo(monitorReq, false);
            if (hostInfos == null || hostInfos.size() == 0) {
                continue;
            }
            Map<String, String> hostMap = new HashMap<>();
            for (ClusterHostModel clusterHostModel : hostInfos) {
                hostMap.put(clusterHostModel.getHostName(), clusterHostModel.getIp());
            }

            String clusterFlag = monitorBaseModel.getClusterFlag();
            String[] clusterFlags = clusterFlag.split(",");

            Map<String, Object> dealHostMap = new HashMap();
            dealHostMap.put("centerId", monitorBaseModel.getCenterId());
            dealHostMap.put("cmFlag", monitorBaseModel.getCmFlag());

            for (int k = 0; k < clusterFlags.length; k++) {

                monitorReq.setClusterFlag(clusterFlags[k]);
                dealHostMap.put("clusterFlag", clusterFlags[k]);

                JSONArray allRolesInfo = clusteAllRolesService.getAllRolesInfo(monitorReq, false);
                for (int e = 0; e < allRolesInfo.size(); e++) {
                    String name = allRolesInfo.getJSONObject(e).getString("name");
                    String serviceType = allRolesInfo.getJSONObject(e).getString("type");
                    if (name.toUpperCase().contains("SPARK")) {
                        continue;
                    }
                    dealHostMap.put("type", serviceType.toUpperCase());
                    dealHostMap.put("name", name.toUpperCase());

                    monitorReq.setMonitorType(name);

                    JSONArray hostRoleInfo = clusterHostRoleService.getHostRoleInfo(monitorReq, false);

                    List<Map<String, String>> hostRoleList = new ArrayList<>();
                    for (int g = 0; g < hostRoleInfo.size(); g++) {
                        JSONObject jsonObject = hostRoleInfo.getJSONObject(g);
                        String type = jsonObject.getString("type");
                        String hostname = jsonObject.getJSONObject("hostRef").getString("hostname");

                        Map<String, String> roleMap = new HashMap<>();
                        roleMap.put("hostJob", dealHostJob(type));
                        roleMap.put("hostName", hostname);
                        roleMap.put("hostIp", hostMap.get(hostname));
                        hostRoleList.add(roleMap);
                    }
                    dealHostMap.put("hostInfoList", hostRoleList);
                    int i1 = saveHostInfo(dealHostMap);
                    totalNum += i1;
                }
            }
        }
        returnMap.put("totalNum", totalNum);
        return returnMap;
    }

    private String dealHostJob(String roleType) {
        switch (roleType) {
            case "NODEMANAGER":
                return "NodeManager";
            case "RESOURCEMANAGER":
                return "ResourceManager";
            case "STATESTORE":
                return "Impala StateStore";
            case "IMPALAD":
                return "Impala Daemon";
            case "NAMENODE":
                return "NameNode";
        }
        return roleType;
    }

    public List<Map<String, String>> selUserUnit(MonitorBaseModel monitorBaseModel) {
        return clusterWhDao.selUserUnit(monitorBaseModel);
    }

}
